<?php

namespace App\Commons\Libs\KafkaPHP;

use Prophecy\Exception\Doubler\MethodNotFoundException;

class Consumer
{
    private $config;

    public function __construct($config)
    {
        $this->config = $config;
    }

    public function start(string $group, array $topics)
    {
        dump($group, $topics);
        $instance = \Kafka\ConsumerConfig::getInstance();
        $instance->setMetadataRefreshIntervalMs($this->config['metadata_refresh_interval_ms']);
        $instance->setMetadataBrokerList($this->config['metadata_broker_list']);
        $instance->setBrokerVersion($this->config['broker_version']);
        // 消费模块的分组ID
        $instance->setGroupId($group);
        // 将要消费的kafka topic名称
        $instance->setTopics($topics);
        // 如果消费offset失效的时候重置offset的策略
        $instance->setOffsetReset('latest');
        $consumer = new \Kafka\Consumer();
        // $consumer->setLogger(logs('kafka-c'));
        $consumer->start(function ($topic, $part, $message) {
            dump($topic, $part, $message);
        });
    }
}
